// PatternMiningMapper.java
package com.taxi.analysis.pattern;

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * 出行模式挖掘Mapper
 * 功能：识别和提取各种出行模式，应用去重技术
 * 技术：去重技术
 */
public class PatternMiningMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Text outputKey = new Text();
    private Text outputValue = new Text();
    private Set<String> processedExactKeys = new HashSet<>();
    private Set<String> processedFuzzyKeys = new HashSet<>();
    private Set<String> processedTimeWindowKeys = new HashSet<>();

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        if (key.get() == 0) return;

        try {
            String line = value.toString().trim();
            if (line.isEmpty()) return;

            String[] fields = parseCSVLine(line);

            if (fields.length >= 23) {
                String pickupDatetime = fields[0].trim();
                String pickupLocationId = fields[1].trim();
                String dropoffLocationId = fields[2].trim();
                int pickupHour = parseInt(fields[3]);
                int pickupWeekday = parseInt(fields[4]);
                String pickupDate = fields[5].trim();
                double tripDistance = parseDouble(fields[6]);
                double totalAmount = parseDouble(fields[7]);
                double tripDuration = parseDouble(fields[8]);
                int passengerCount = parseInt(fields[9]);
                String routePattern = fields[10].trim();
                String timePattern = fields[11].trim();
                String behaviorPattern = fields[12].trim();
                String compositePattern = fields[13].trim();
                String userPatternId = fields[14].trim();
                int routeFrequency = parseInt(fields[15]);
                int timeFrequency = parseInt(fields[16]);
                int behaviorFrequency = parseInt(fields[17]);
                int compositeFrequency = parseInt(fields[18]);
                double patternImportance = parseDouble(fields[19]);
                String patternType = fields[20].trim();
                String exactDedupKey = fields[21].trim();
                String fuzzyDedupKey = fields[22].trim();
                String timeWindowKey = fields[23].trim();

                if (isValidData(pickupLocationId, dropoffLocationId)) {

                    // 去重技术1: 精确去重
                    processExactDeduplication(exactDedupKey, pickupLocationId, dropoffLocationId,
                            pickupDatetime, routePattern, timePattern, behaviorPattern,
                            compositePattern, tripDistance, totalAmount, tripDuration,
                            patternImportance, patternType, context);

                    // 去重技术2: 模糊去重
                    processFuzzyDeduplication(fuzzyDedupKey, pickupLocationId, dropoffLocationId,
                            pickupHour, pickupWeekday, routePattern, timePattern,
                            tripDistance, totalAmount, routeFrequency, timeFrequency,
                            patternImportance, context);

                    // 去重技术3: 时间窗口去重
                    processTimeWindowDeduplication(timeWindowKey, routePattern, pickupDate,
                            pickupHour, tripDistance, totalAmount, tripDuration,
                            routeFrequency, patternImportance, context);

                    // 去重技术4: 用户模式去重
                    processUserPatternDeduplication(userPatternId, routePattern, timePattern,
                            behaviorPattern, pickupLocationId, tripDistance,
                            totalAmount, patternImportance, patternType, context);
                }
            }

        } catch (Exception e) {
            System.err.println("处理行时出错: " + value.toString());
        }
    }

    /**
     * 精确去重处理
     */
    private void processExactDeduplication(String exactKey, String pickupLocationId, String dropoffLocationId,
                                           String pickupDatetime, String routePattern, String timePattern,
                                           String behaviorPattern, String compositePattern, double tripDistance,
                                           double totalAmount, double tripDuration, double patternImportance,
                                           String patternType, Context context)
            throws IOException, InterruptedException {

        // 精确去重：只处理第一次出现的完全相同记录
        if (!processedExactKeys.contains(exactKey)) {
            processedExactKeys.add(exactKey);

            String dedupKey = "EXACT_DEDUP_" + exactKey;
            String exactRecord = createExactRecord(pickupLocationId, dropoffLocationId, pickupDatetime,
                    routePattern, timePattern, behaviorPattern, compositePattern,
                    tripDistance, totalAmount, tripDuration, patternImportance, patternType);

            outputKey.set(dedupKey);
            outputValue.set("EXACT_PATTERN|" + exactRecord);
            context.write(outputKey, outputValue);
        }
    }

    /**
     * 模糊去重处理
     */
    private void processFuzzyDeduplication(String fuzzyKey, String pickupLocationId, String dropoffLocationId,
                                           int pickupHour, int pickupWeekday, String routePattern, String timePattern,
                                           double tripDistance, double totalAmount, int routeFrequency,
                                           int timeFrequency, double patternImportance, Context context)
            throws IOException, InterruptedException {

        // 模糊去重：合并相似的记录
        if (!processedFuzzyKeys.contains(fuzzyKey)) {
            processedFuzzyKeys.add(fuzzyKey);

            String dedupKey = "FUZZY_DEDUP_" + fuzzyKey;
            String fuzzyRecord = createFuzzyRecord(pickupLocationId, dropoffLocationId, pickupHour,
                    pickupWeekday, routePattern, timePattern, tripDistance,
                    totalAmount, routeFrequency, timeFrequency, patternImportance);

            outputKey.set(dedupKey);
            outputValue.set("FUZZY_PATTERN|" + fuzzyRecord);
            context.write(outputKey, outputValue);
        }
    }

    /**
     * 时间窗口去重处理
     */
    private void processTimeWindowDeduplication(String timeWindowKey, String routePattern, String pickupDate,
                                                int pickupHour, double tripDistance, double totalAmount,
                                                double tripDuration, int routeFrequency, double patternImportance,
                                                Context context)
            throws IOException, InterruptedException {

        // 时间窗口去重：同一时间窗口内的相似记录只保留一个代表
        if (!processedTimeWindowKeys.contains(timeWindowKey)) {
            processedTimeWindowKeys.add(timeWindowKey);

            String dedupKey = "TIME_WINDOW_" + timeWindowKey;
            String timeWindowRecord = createTimeWindowRecord(routePattern, pickupDate, pickupHour,
                    tripDistance, totalAmount, tripDuration,
                    routeFrequency, patternImportance);

            outputKey.set(dedupKey);
            outputValue.set("TIME_WINDOW_PATTERN|" + timeWindowRecord);
            context.write(outputKey, outputValue);
        }
    }

    /**
     * 用户模式去重处理
     */
    private void processUserPatternDeduplication(String userPatternId, String routePattern, String timePattern,
                                                 String behaviorPattern, String pickupLocationId, double tripDistance,
                                                 double totalAmount, double patternImportance, String patternType,
                                                 Context context)
            throws IOException, InterruptedException {

        String dedupKey = "USER_PATTERN_" + userPatternId;
        String userRecord = createUserPatternRecord(routePattern, timePattern, behaviorPattern,
                pickupLocationId, tripDistance, totalAmount,
                patternImportance, patternType);

        outputKey.set(dedupKey);
        outputValue.set("USER_PATTERN|" + userRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 创建精确记录
     */
    private String createExactRecord(String pickupLocationId, String dropoffLocationId, String pickupDatetime,
                                     String routePattern, String timePattern, String behaviorPattern,
                                     String compositePattern, double tripDistance, double totalAmount,
                                     double tripDuration, double patternImportance, String patternType) {

        return String.format("%s|%s|%s|%s|%s|%s|%s|%.2f|%.2f|%.2f|%.3f|%s",
                pickupLocationId, dropoffLocationId, pickupDatetime, routePattern,
                timePattern, behaviorPattern, compositePattern, tripDistance,
                totalAmount, tripDuration, patternImportance, patternType);
    }

    /**
     * 创建模糊记录
     */
    private String createFuzzyRecord(String pickupLocationId, String dropoffLocationId, int pickupHour,
                                     int pickupWeekday, String routePattern, String timePattern,
                                     double tripDistance, double totalAmount, int routeFrequency,
                                     int timeFrequency, double patternImportance) {

        return String.format("%s|%s|%d|%d|%s|%s|%.2f|%.2f|%d|%d|%.3f",
                pickupLocationId, dropoffLocationId, pickupHour, pickupWeekday,
                routePattern, timePattern, tripDistance, totalAmount,
                routeFrequency, timeFrequency, patternImportance);
    }

    /**
     * 创建时间窗口记录
     */
    private String createTimeWindowRecord(String routePattern, String pickupDate, int pickupHour,
                                          double tripDistance, double totalAmount, double tripDuration,
                                          int routeFrequency, double patternImportance) {

        return String.format("%s|%s|%d|%.2f|%.2f|%.2f|%d|%.3f",
                routePattern, pickupDate, pickupHour, tripDistance,
                totalAmount, tripDuration, routeFrequency, patternImportance);
    }

    /**
     * 创建用户模式记录
     */
    private String createUserPatternRecord(String routePattern, String timePattern, String behaviorPattern,
                                           String pickupLocationId, double tripDistance, double totalAmount,
                                           double patternImportance, String patternType) {

        return String.format("%s|%s|%s|%s|%.2f|%.2f|%.3f|%s",
                routePattern, timePattern, behaviorPattern, pickupLocationId,
                tripDistance, totalAmount, patternImportance, patternType);
    }

    private boolean isValidData(String pickupLocationId, String dropoffLocationId) {
        return !pickupLocationId.isEmpty() && !dropoffLocationId.isEmpty() &&
                !pickupLocationId.equals("null") && !dropoffLocationId.equals("null");
    }

    private String[] parseCSVLine(String line) {
        if (line.contains("\"")) {
            return parseQuotedCSV(line);
        } else {
            return line.split(",");
        }
    }

    private String[] parseQuotedCSV(String line) {
        List<String> fields = new ArrayList<>();
        boolean inQuotes = false;
        StringBuilder currentField = new StringBuilder();

        for (int i = 0; i < line.length(); i++) {
            char c = line.charAt(i);

            if (c == '"') {
                inQuotes = !inQuotes;
            } else if (c == ',' && !inQuotes) {
                fields.add(currentField.toString());
                currentField = new StringBuilder();
            } else {
                currentField.append(c);
            }
        }
        fields.add(currentField.toString());

        return fields.toArray(new String[0]);
    }

    private double parseDouble(String str) {
        try {
            return Double.parseDouble(str.trim());
        } catch (NumberFormatException e) {
            return 0.0;
        }
    }

    private int parseInt(String str) {
        try {
            return Integer.parseInt(str.trim());
        } catch (NumberFormatException e) {
            return 0;
        }
    }
}
